package com.vincent.leader;

import com.vincent.common.Command;
import com.vincent.common.Environment;
import com.vincent.common.Message;
import com.vincent.common.Process;
import com.vincent.message.DecisionMessage;
import com.vincent.message.P2aMessage;
import com.vincent.message.P2bMessage;
import com.vincent.message.PreemptedMessage;
import com.vincent.utils.BallotNumber;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

/**
 * Commander运行在Synod协议的第二阶段()
 *
 * Created by wensen on 2016/12/11.
 */
public class Commander extends Process{

    private static Logger LOGGER = LoggerFactory.getLogger(Commander.class);

//    private Environment env;
//    private int id;
    private final int leader;
    private final List<Integer> acceptors;
    private final List<Integer> replicas;
    private final BallotNumber ballot_number;
    private final long slot_number;
    private final Command command;

    /**
     * Builder模式，参考<<Effective Java>>第二条
     */
    public static class Builder{

        private Environment env;
        private int id;
        private int leader;
        private List<Integer> acceptors;
        private List<Integer> replicas;
        private BallotNumber ballot_number;
        private long slot_number;
        private Command command;

        public Builder() {}

        public Builder env(Environment env) {
            env = env;
            return this;
        }

        public Builder id(int id) {
            id = id;
            return this;
        }

        public Builder leader(int leader) {
            leader = leader;
            return this;
        }

        public Builder acceptors(List<Integer> as) {
            acceptors = new ArrayList<Integer>();
            acceptors.addAll(as);
            return this;
        }

        public Builder replicas(List<Integer> rs) {
            replicas = new ArrayList<Integer>();
            replicas.addAll(rs);
            return this;
        }

        public Builder ballot_number(BallotNumber ballot) {

            ballot_number = new BallotNumber(ballot.round(), ballot.leader_id());
            return this;
        }

        public Builder slot_number(long slot) {

            slot_number = slot;
            return this;
        }

        public Builder command(Command command) {
            command = command;
            return this;
        }

        public Commander build() {
            return new Commander(this);
        }

    }

    private Commander(Builder builder) {
        super(builder.env, builder.id);
        leader        = builder.leader;
        acceptors     = builder.acceptors;
        replicas      = builder.replicas;
        ballot_number = builder.ballot_number;
        slot_number   = builder.slot_number;
        command       = builder.command;
        this.addProc(this);
    }

    /**
     * commander发送p2a信息到所有acceptors，并等待p2b回应。回复信息的ballot将大于
     * commander的ballot,这里有两种情况：
     *
     * 1、如果commander接收的p2b信息来自大多数(过半)acceptor,那么commander学习到
     *      该command已经在slot中所确定，那么它将通知所有的replica并退出
     *
     * 2. 如果commander接收到的p2b信息中，有些ballot不同于自身的ballot，那么它知道
     *      一个更大的ballot被accepter所接受，这就意味着commander自身的这个ballot
     *      不能再使用，在这种情况下，commander将提醒它的leader有更高的ballot存在，
     *      并退出
     */
    public void body() {

        List<Integer> waitfor = new LinkedList<Integer>();
        //发送p2a信息给所有acceptors,并加入等待
        for (Integer a : this.acceptors) {

            this.sendMessage(a, new P2aMessage(this.id(), this.ballot_number,
                    this.slot_number, this.command));
            waitfor.add(a);
        }

        while (true) {

            Message msg = null;
            try {
                msg = this.getNextMessage();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            if (msg instanceof P2bMessage) {
                if (this.ballot_number == ((P2bMessage) msg).ballot_number()
                        && waitfor.contains(msg.src())) {
                    waitfor.remove(msg.src());
                    //过半时，说明已经确定值(command)
                    if (waitfor.size() < this.acceptors.size()/2) {
                        for (Integer r : this.replicas) {
                            this.sendMessage(r, new DecisionMessage(this.id(), this.slot_number,
                                    this.command));
                        }
                    }
                    /**
                     * 当有更高的ballot时
                     */
                } else {
                    LOGGER.debug("in p2b message, there has a greater ballot: {}",
                            ((P2bMessage) msg).ballot_number());
                    this.sendMessage(this.leader, new PreemptedMessage(this.id(),
                            ((P2bMessage) msg).ballot_number()));
                }
            } else {
                LOGGER.debug("recv msg is not P2bMessage, it is {}", msg);
            }
        }
    }
}
